package cn.itcast.flink.transformation;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MinMinByDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple3<Integer, Integer, Integer>> source = env.fromElements(
                Tuple3.of(1, 3, 2),
                Tuple3.of(1, 1, 2),
                Tuple3.of(1, 2, 3),
                Tuple3.of(1, 111, 1),
                Tuple3.of(1, 1, 1),
                Tuple3.of(1, 2, 0),
                Tuple3.of(1, 33, 2)
        );
        //取最小值，别的值按顺序填充
        source.keyBy(t -> t.f0).min(2).print("min>>>");
        //获取最小值，取第一个最小值的整行数据
        source.keyBy(t -> t.f0).minBy(2).printToErr("minBy>>>");

        /*
         * min 会取最小值记录下来, 其它字段按照顺序填充, 最终将最后一个元素的其它字段和最小值组合形成结果返回
         * minBy 会判断最小值, 将第一个出现的最小值所在的那一行的结果保留, 最终返回出去
         * max maxBy同理
         */
        env.execute();
    }
}